package pl.edu.agh.mr.nb.server;

import java.util.ArrayList;
import java.util.Properties;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import pl.edu.agh.mr.nb.node.ComputationOrderMsg;
import pl.edu.agh.mr.nb.node.ComputationResultMsg;
import pl.edu.agh.mr.nb.node.NodeAction;
import pl.edu.agh.mr.nb.node.ObjMessage;
import sun.misc.Cleaner;
import cgl.narada.jms.NBJmsInitializer;


public class FractalServer implements MessageListener{	

	private TopicSession registeringPubSession;
    private TopicConnection connection;
    private String nodeToServerTopic = "nodeToServerTopic";
    private String clientToServerTopic = "clientToServerTopic";
    private NodeArray nodesList;
    private ClientArray clientList;
    
	private TopicSubscriber registeringSubscriber;
	private TopicSession nodeResultsSession;
	private TopicSubscriber resultsSubscriber;
	private TopicSession clientRequestSession;
	private TopicSubscriber ordersSubscriber;
      
    public static boolean isRunning = true;
    
    private int divideRatio;

    public FractalServer(Properties connectionProperties, String transportType, int divideRatio)  throws Exception {

    	this.divideRatio = divideRatio;
    	
    	NBJmsInitializer ini;
        ini = new NBJmsInitializer(connectionProperties, transportType);

        // Lookup a JMS connection factory
        TopicConnectionFactory conFactory = (TopicConnectionFactory) ini.lookup();
        initializeSession(conFactory);
        nodesList = new NodeArray();
        clientList = new ClientArray();
        
        System.out.println("Server is working and ready for requests...");
    }

    private void initializeSession(TopicConnectionFactory conFactory) throws Exception {

    	
        // Create a JMS connection
        TopicConnection connection = conFactory.createTopicConnection("guest", "password");

        this.connection = connection; 

       
        // subscriber for node registering 
        registeringPubSession = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
        Topic registeringTopic = registeringPubSession.createTopic(nodeToServerTopic+"/register");
        registeringSubscriber = registeringPubSession.createSubscriber(registeringTopic);
        registeringSubscriber.setMessageListener(this); //setting message listener
    

        // communication from Node to Server
        nodeResultsSession = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
        Topic resultsTopic = nodeResultsSession.createTopic(nodeToServerTopic+"/results");
        resultsSubscriber = nodeResultsSession.createSubscriber(resultsTopic);
        resultsSubscriber.setMessageListener(this);
        
        //communication from Client to Server 
        clientRequestSession = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
        Topic clientRequestTopic = clientRequestSession.createTopic(clientToServerTopic);
        ordersSubscriber = clientRequestSession.createSubscriber(clientRequestTopic);
        ordersSubscriber.setMessageListener(this);
        
        connection.start(); // starting connection
    }
   

//===================================================================================
 /* Run the Fractal Server  */
    public static void main(String[] args) {
        
    	String transport = "niotcp";
        
        Runtime.getRuntime().addShutdownHook(new Thread() {
        	public void run() {
        		isRunning = false;
        	}
        });

        try {
            if ( args.length != 3) {
                System.out.println("Usage: java pl.edu.agh.mr.nb.server.FractalServer <hostname> " + "<portNum> " + "[divideRatio]");
                System.exit(0);
            }

            Properties props = new Properties();

            /** These properties pertain to setting up a TCP link */
            props.put("hostname", args[0]);
            props.put("portnum", args[1]);

            int divideRatio = 10;
            
            if (args.length == 3) {
            	divideRatio = Integer.decode(args[2]);
            }
            
            FractalServer server = new FractalServer(props, transport, divideRatio);

      
    
         } catch (Exception e){
            e.printStackTrace();
        }

    }

    //sending an order to idle node 
    private void sendOrderToNode(ComputationOrderMsg order) {
		
    	while (true){
    		Node idleNode = nodesList.getIdleNode();
    		
    		if (idleNode != null){
    			// if there is any free node
    			
    			try {
				
    				idleNode.sendOrder(order); // sending order to the computation node
    				break;
    				
    			} catch (JMSException e) {
					e.printStackTrace();
				}
    			
    		}
    	}
		
	}

	// just for tests whether onMessage works
    // function is responisble for asynchronious message receiving and taking appropriate actions 
	public void onMessage(Message msg) {

	System.out.println("Got the message");
	if (msg != null){
	
		if (msg instanceof TextMessage){
		
			TextMessage tmsg = (TextMessage) msg;
			
			try {
				
				// if received message is from node 
				if (tmsg.getIntProperty("type") == ObjMessage.NODE){
					if (tmsg.getIntProperty("action") == NodeAction.REGISTER_NODE){
						registerNode(tmsg.getText());		        				
					}
					
					if (tmsg.getIntProperty("action") == NodeAction.UNREGISTER_NODE){
						unregisterNode(tmsg.getText());
					}
				}
			
				// if received message is from client
				if (tmsg.getIntProperty("type") == ObjMessage.CLIENT){

					if (tmsg.getIntProperty("action") == NodeAction.REGISTER_NODE){
						registerClient(tmsg.getText());		        				
					}
						
					if (tmsg.getIntProperty("action") == NodeAction.UNREGISTER_NODE){
						unregisterClient(tmsg.getText());
					}
				
				}
			} catch (JMSException e) {
				e.printStackTrace();
			}
			
		}
		else if (msg instanceof ObjectMessage) {
			
			final ObjectMessage omsg = (ObjectMessage)msg;
			
			try {
				
				if (omsg.getIntProperty("type") == ObjMessage.NODE){
					System.out.println("Got message from node");
					handleNodeResults(omsg);
				}else if (omsg.getIntProperty("type") == ObjMessage.CLIENT){
					System.out.println("Got an order from client");
					new Thread() {
						public void run() {
							handleClientOrder(omsg);
						}
					}.start();
				}
				
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			}
		
		}
		System.out.println("Leaving onMessage()");
	}

	private void unregisterClient(String clientName) {
		// TODO Auto-generated method stub
		clientList.unregisterClient(clientName);
	}

	private void registerClient(String clientName) throws JMSException {
		TopicSession session = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
		Client client = new Client(clientName,session,clientToServerTopic,divideRatio);
		clientList.registerClient(client);
	}

	//handling node results and sending them back to 
	private void handleNodeResults(ObjectMessage omsg) {
		
		try {
			
			ComputationResultMsg result = (ComputationResultMsg)omsg.getObject();

			Client client = clientList.getClientByName(result.clientName);
			
			// adding part of the results to client 
			client.appendPartlyResults(result);
			
			// checking if the client has coplete results 
			if (client.readyToSendBack()){
				client.sendResultsToClient();	    
			}
			
		
			// marking the node as free, i know it may be pretty confusing but 
			// it need a better implementation 
			// by now it's a kind of workaround
			
			Node node = nodesList.getNodeByName(omsg.getStringProperty("nodeName"));
			if (node == null){	
				System.out.println("Something is wrong ... The Node should exist");
			}
			else{
				nodesList.registerNode(node); // once again the node is working 
			}
				
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	// function handling client request 
	private void handleClientOrder(ObjectMessage object) {
		
		System.out.println("HandleClientOrder");
		
		
		try {
			
			ComputationOrderMsg order = (ComputationOrderMsg) object.getObject();
			
			Client client = clientList.getClientByName(order.clientName);
				
			if (client==null){
				System.out.println("There is no such client "+order.clientName);
			}
 
			// the client order is splited into smaller orders and sended one by one 
			ArrayList<ComputationOrderMsg> ordersList = splitOrder(order);
			
			for (int i=0; i< ordersList.size(); i++)
				sendOrderToNode(ordersList.get(i));
			//----------------------------------------------------------------------
		
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	//unregistering node 
	private void unregisterNode(String nodeName) {	
		nodesList.unregisterNode(nodeName);
		System.out.println("Node "+nodeName+" unregistered...");
	}

	// registering node Name 
	private void registerNode(String nodeName) {
		
        TopicSession session;
		try {
			session = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
			Node node = new Node(nodeName,session,nodeToServerTopic);
			nodesList.registerNode(node);
			System.out.println("Node "+nodeName+" registered...");

		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	private ArrayList<ComputationOrderMsg> splitOrder(ComputationOrderMsg order){
		ArrayList<ComputationOrderMsg> parts = new ArrayList<ComputationOrderMsg>(divideRatio);
		
		int partSize = order.ny/divideRatio;
		double dy = (order.y2 - order.y1)/order.ny;
		
		for(int i=0; i < divideRatio; i++) {
			ComputationOrderMsg part = new ComputationOrderMsg(order);
			
			// marking each part for each client 
			part.clientName = order.clientName;
			part.id = i;
						
			part.y1 = order.y1 + i*dy*partSize;
			part.y2 = order.y1 + (i+1)*dy*partSize;
			part.ny = partSize;
			parts.add(part);
		}
		parts.get(divideRatio-1).y2 = order.y2;
		parts.get(divideRatio-1).ny = order.ny - (divideRatio-1)*partSize;
		
		return parts;
	}

}


